// 
// Copyright (C) 2000-2003
// magustek co.
//
#ifndef MAGUS_CONDITION_IMPL_HPP
#define MAGUS_CONDITION_IMPL_HPP

#include <magus/thread/xtime.hpp>
#include <magus/thread/thread.hpp>
#include <magus/thread/exceptions.hpp>
//#include <magus/limits.hpp>
#include <climits>
#include <cassert>

#if defined(MAGUS_HAS_WINTHREADS)
#   ifndef NOMINMAX
#      define NOMINMAX
#   endif
#   include <windows.h>
#elif defined(MAGUS_HAS_PTHREADS)
#   include <errno.h>
#endif

namespace magus {

namespace detail {

class MAGUS_THREAD_DECL condition_impl : private noncopyable
{
    friend class condition;

public:
#if defined(MAGUS_HAS_WINTHREADS)
    void* m_gate;
    void* m_queue;
    void* m_mutex;
    unsigned m_gone; 			// # threads that timed out and never made it to m_queue
    unsigned long m_blocked; 	// # threads blocked on the condition
    unsigned m_waiting; 		// # threads no longer waiting for the condition but
                        		// still waiting to be removed from m_queue
#elif defined(MAGUS_HAS_PTHREADS)
    pthread_cond_t m_condition;
#endif

#if defined(MAGUS_HAS_WINTHREADS)
	condition_impl()
	    : m_gone(0), m_blocked(0), m_waiting(0)
	{
	    m_gate = reinterpret_cast<void*>(CreateSemaphore(0, 1, 1, 0));
	    m_queue = reinterpret_cast<void*>(CreateSemaphore(0, 0, LONG_MAX, 0));
	              //std::numeric_limits<long>::max(), 0));
	    m_mutex = reinterpret_cast<void*>(CreateMutex(0, 0, 0));
	
	    if (!m_gate || !m_queue || !m_mutex)
	        {
	            int res = 0;
	            if (m_gate)
	                {
	                    res = CloseHandle(reinterpret_cast<HANDLE>(m_gate));
	                    assert(res);
	                }
	            if (m_queue)
	                {
	                    res = CloseHandle(reinterpret_cast<HANDLE>(m_queue));
	                    assert(res);
	                }
	            if (m_mutex)
	                {
	                    res = CloseHandle(reinterpret_cast<HANDLE>(m_mutex));
	                    assert(res);
	                }
	
	            throw thread_resource_error();
	        }
	}
	
	~condition_impl()
	{
	    int res = 0;
	    res = CloseHandle(reinterpret_cast<HANDLE>(m_gate));
	    assert(res);
	    res = CloseHandle(reinterpret_cast<HANDLE>(m_queue));
	    assert(res);
	    res = CloseHandle(reinterpret_cast<HANDLE>(m_mutex));
	    assert(res);
	}
	
	void notify_one()
	{
	    unsigned signals = 0;
	
	    int res = 0;
	    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_mutex), INFINITE);
	    assert(res == WAIT_OBJECT_0);
	
	    if (m_waiting != 0) // the m_gate is already closed
	        {
	            if (m_blocked == 0)
	                {
	                    res = ReleaseMutex(reinterpret_cast<HANDLE>(m_mutex));
	                    assert(res);
	                    return;
	                }
	
	            ++m_waiting;
	            --m_blocked;
	            signals = 1;
	        }
	    else
	        {
	            res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_gate), INFINITE);
	            assert(res == WAIT_OBJECT_0);
	            if (m_blocked > m_gone)
	                {
	                    if (m_gone != 0)
	                        {
	                            m_blocked -= m_gone;
	                            m_gone = 0;
	                        }
	                    signals = m_waiting = 1;
	                    --m_blocked;
	                }
	            else
	                {
	                    res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1, 0);
	                    assert(res);
	                }
	        }
	
	    res = ReleaseMutex(reinterpret_cast<HANDLE>(m_mutex));
	    assert(res);
	
	    if (signals)
	        {
	            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_queue), signals, 0);
	            assert(res);
	        }
	}
	
	void notify_all()
	{
	    unsigned signals = 0;
	
	    int res = 0;
	    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_mutex), INFINITE);
	    assert(res == WAIT_OBJECT_0);
	
	    if (m_waiting != 0) // the m_gate is already closed
	        {
	            if (m_blocked == 0)
	                {
	                    res = ReleaseMutex(reinterpret_cast<HANDLE>(m_mutex));
	                    assert(res);
	                    return;
	                }
	
	            m_waiting += (signals = m_blocked);
	            m_blocked = 0;
	        }
	    else
	        {
	            res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_gate), INFINITE);
	            assert(res == WAIT_OBJECT_0);
	            if (m_blocked > m_gone)
	                {
	                    if (m_gone != 0)
	                        {
	                            m_blocked -= m_gone;
	                            m_gone = 0;
	                        }
	                    signals = m_waiting = m_blocked;
	                    m_blocked = 0;
	                }
	            else
	                {
	                    res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1, 0);
	                    assert(res);
	                }
	        }
	
	    res = ReleaseMutex(reinterpret_cast<HANDLE>(m_mutex));
	    assert(res);
	
	    if (signals)
	        {
	            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_queue), signals,
	                                   0);
	            assert(res);
	        }
	}
	
	void enter_wait()
	{
	    int res = 0;
	    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_gate), INFINITE);
	    assert(res == WAIT_OBJECT_0);
	    ++m_blocked;
	    res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1, 0);
	    assert(res);
	}
	
	void do_wait()
	{
	    int res = 0;
	    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_queue), INFINITE);
	    assert(res == WAIT_OBJECT_0);
	
	    unsigned long was_waiting=0;
	    unsigned long was_gone=0;
	
	    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_mutex), INFINITE);
	    assert(res == WAIT_OBJECT_0);
	    was_waiting = m_waiting;
	    was_gone = m_gone;
	    if (was_waiting != 0)
	        {
	            if (--m_waiting == 0)
	                {
	                    if (m_blocked != 0)
	                        {
	                            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1,
	                                                   0); // open m_gate
	                            assert(res);
	                            was_waiting = 0;
	                        }
	                    else if (m_gone != 0)
	                        m_gone = 0;
	                }
	        }
	    else if (++m_gone == (LONG_MAX / 2)) // (std::numeric_limits<unsigned>::max() / 2))
	        {
	            // timeout occured, normalize the m_gone count
	            // this may occur if many calls to wait with a timeout are made and
	            // no call to notify_* is made
	            res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_gate), INFINITE);
	            assert(res == WAIT_OBJECT_0);
	            m_blocked -= m_gone;
	            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1, 0);
	            assert(res);
	            m_gone = 0;
	        }
	    res = ReleaseMutex(reinterpret_cast<HANDLE>(m_mutex));
	    assert(res);
	
	    if (was_waiting == 1)
	        {
	            for ( ; was_gone; --was_gone)
	                {
	                    // better now than spurious later
	                    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_queue),
	                                              INFINITE);
	                    assert(res == WAIT_OBJECT_0);
	                }
	            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1, 0);
	            assert(res);
	        }
	}
	
	bool do_timed_wait(const xtime& xt)
	{
	    bool ret = false;
	    unsigned int res = 0;
	
	    for (;;)
	        {
	            int milliseconds;
	            to_duration(xt, milliseconds);
	
	            res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_queue),
	                                      milliseconds);
	            assert(res != WAIT_FAILED && res != WAIT_ABANDONED);
	            ret = (res == WAIT_OBJECT_0);
	
	            if (res == WAIT_TIMEOUT)
	                {
	                    xtime cur;
	                    xtime_get(&cur, TIME_UTC);
	                    if (xtime_cmp(xt, cur) > 0)
	                        continue;
	                }
	
	            break;
	        }
	
	    unsigned was_waiting=0;
	    unsigned was_gone=0;
	
	    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_mutex), INFINITE);
	    assert(res == WAIT_OBJECT_0);
	    was_waiting = m_waiting;
	    was_gone = m_gone;
	    if (was_waiting != 0)
	        {
	            if (!ret) // timeout
	                {
	                    if (m_blocked != 0)
	                        --m_blocked;
	                    else
	                        ++m_gone; // count spurious wakeups
	                }
	            if (--m_waiting == 0)
	                {
	                    if (m_blocked != 0)
	                        {
	                            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1,
	                                                   0); // open m_gate
	                            assert(res);
	                            was_waiting = 0;
	                        }
	                    else if (m_gone != 0)
	                        m_gone = 0;
	                }
	        }
	    else if (++m_gone == (LONG_MAX / 2)) // (std::numeric_limits<unsigned>::max() / 2))
	        {
	            // timeout occured, normalize the m_gone count
	            // this may occur if many calls to wait with a timeout are made and
	            // no call to notify_* is made
	            res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_gate), INFINITE);
	            assert(res == WAIT_OBJECT_0);
	            m_blocked -= m_gone;
	            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1, 0);
	            assert(res);
	            m_gone = 0;
	        }
	    res = ReleaseMutex(reinterpret_cast<HANDLE>(m_mutex));
	    assert(res);
	
	    if (was_waiting == 1)
	        {
	            for ( ; was_gone; --was_gone)
	                {
	                    // better now than spurious later
	                    res = WaitForSingleObject(reinterpret_cast<HANDLE>(m_queue),
	                                              INFINITE);
	                    assert(res ==  WAIT_OBJECT_0);
	                }
	            res = ReleaseSemaphore(reinterpret_cast<HANDLE>(m_gate), 1, 0);
	            assert(res);
	        }
	
	    return ret;
	}
	
//////////////////////////////////////////////////////////////////////////////
#elif defined(MAGUS_HAS_PTHREADS)
	condition_impl()
	{
	    int res = 0;
	    res = pthread_cond_init(&m_condition, 0);
	    if (res != 0)
	        throw thread_resource_error();
	}
	
	~condition_impl()
	{
	    int res = 0;
	    res = pthread_cond_destroy(&m_condition);
	    assert(res == 0);
	}
	
	void notify_one()
	{
	    int res = 0;
	    res = pthread_cond_signal(&m_condition);
	    assert(res == 0);
	}
	
	void notify_all()
	{
	    int res = 0;
	    res = pthread_cond_broadcast(&m_condition);
	    assert(res == 0);
	}
	
	void do_wait(pthread_mutex_t* pmutex)
	{
	    int res = 0;
	    res = pthread_cond_wait(&m_condition, pmutex);
	    assert(res == 0);
	}
	
	bool do_timed_wait(const xtime& xt, pthread_mutex_t* pmutex)
	{
	    timespec ts;
	    to_timespec(xt, ts);
	
	    int res = 0;
	    res = pthread_cond_timedwait(&m_condition, pmutex, &ts);
	    assert(res == 0 || res == ETIMEDOUT);
	
	    return res != ETIMEDOUT;
	}
#endif // MAGUS_HAS_PTHREADS
	
}; // class condition_impl

} // namespace detail

} // namespace magus

// Change Log:
//    16 JUN 03  Initial version.

#endif // MAGUS_CONDITION_IMPL_HPP
